package com.hzsai.service.impl;

import com.hzsai.kafka.KafkaSender;
import com.hzsai.service.PushService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * @author: hzsai
 * @date : 2020/3/26/ 18:47
 **/
@Service
public class PushServiceImpl implements PushService {

    private static final Logger logger = LoggerFactory.getLogger(PushServiceImpl.class);

    private final KafkaTemplate<String, String> kafkaTemplate;

    private KafkaSender sender;

    public PushServiceImpl(KafkaTemplate kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
        sender = new KafkaSender(kafkaTemplate);
    }

    @Override
    public int push(String message) {
        // TODO:
        // 目前这里只是推动到了kafka，没有进行业务操作
        // 这里，kafka推送到各个socketIOClient作为一个微服务？
        ListenableFuture future = sender.sendAndResult(kafkaTemplate.getDefaultTopic(), message);
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onSuccess(SendResult<String, String> result) {
                logger.info("Kafka send result: {}", result.getProducerRecord());
                logger.info("Kafka message send successfully. --- topic: {} - msg: {}",
                        kafkaTemplate.getDefaultTopic(),
                        message);
            }
            @Override
            public void onFailure(Throwable ex) {
                logger.error("kafka message send failed. -- topic: {} - msg: {}",
                        kafkaTemplate.getDefaultTopic(),
                        message);
            }
        });
        return 0;
    }
}
